package com.simple.mqtt.common.task;

import com.simple.mqtt.common.tool.LogTool;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.util.concurrent.ScheduledExecutorService;
import java.util.function.BiFunction;

/**
 * Abstract base class for MQTT message retransmission
 * Provides unified retransmission mechanism implementation for different MQTT message types
 * 
 * @param <T> message type generic, must extend MqttMessage
 */
public abstract class AbstractMqttPendingMessage<T extends MqttMessage> {

    /** Retransmission processor */
    protected final RetryProcessor<T> retryProcessor;
    /** Quality of Service level */
    protected final MqttQoS qos;

    /**
     * Constructor
     * @param qos quality of service level
     * @param retryProcessor retransmission processor instance
     */
    protected AbstractMqttPendingMessage(MqttQoS qos, RetryProcessor<T> retryProcessor) {
        this.qos = qos;
        this.retryProcessor = retryProcessor;
    }

    /**
     * Start message retransmission timer
     * @param taskService scheduled task service
     * @param context channel context
     * @param messageFactory message factory function for creating retransmission messages
     * @param msgType message type identifier for logging
     */
    public void startRetransmissionTimer(ScheduledExecutorService taskService,
                                         ChannelHandlerContext context,
                                         BiFunction<MqttFixedHeader, T, ? extends MqttMessage> messageFactory,
                                         String msgType) {
        this.retryProcessor.setHandler((fixedHeader, originalMessage) -> {
            MqttMessage message = messageFactory.apply(fixedHeader, originalMessage);
            context.writeAndFlush(message).addListener(future -> {
                if (context.channel().isActive()) {
                    if (context.channel().isWritable()) {
                        LogTool.info("retry send {} msg qos:{} result:{}", msgType, qos, future.isSuccess());
                    } else {
                        LogTool.warn("retry send {} msg qos:{} failed, channel not writable", msgType, qos);
                    }
                } else {
                    LogTool.warn("retry send {} msg qos:{} failed, channel inactive", msgType, qos);
                }
            });
        });
        this.retryProcessor.start(taskService);
    }

    /**
     * Stop message retransmission
     * Usually called after receiving acknowledgment message
     */
    public void stopRetransmission() {
        this.retryProcessor.stop();
    }
}